package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MykafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName DwdTrafficUserNewUser
 * @Description TODO 新增用户事务事实表
 * @Author AFei
 * @Date 2022/9/5 19:44
 * @Version 1.0
 */
public class DwdTrafficUserNewUser {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 3. 从topic_db主题中读取业务表变化数据 创建动态表
        tableEnv.executeSql(MykafkaUtil.getTopicDDL("dwd_traffic_user_new_add_group"));

        //tableEnv.executeSql("select * from topic_db").print();

        // TODO 4. 从业务数据中筛选出新增用户数据
        Table userInfo = tableEnv.sqlQuery("select \n" +
                "                data['id'] id, \n" +
                "                data['login_name'] login_name, \n" +
                "                data['phone_num'] phone_num, \n" +
                "                data['email'] email,\n" +
                "                data['birthday'] birthday, \n" +
                "                ts \n" +
                "            from topic_db where `table`='user_info' and \n" +
                "             `type` = 'insert'");
        tableEnv.createTemporaryView("user_info",userInfo);
        //tableEnv.executeSql("select * from user_info").print();

        // TODO 5. 创建动态表和要写入的kafka主题进行映射

        tableEnv.executeSql("create table dwd_traffic_user_new_user(\n" +
                "            id string,\n" +
                "            login_name string,\n" +
                "            phone_num string,\n" +
                "            email string,\n" +
                "            birthday string,\n" +
                "            ts string,\n" +
                "            primary key(id) not enforced\n" +
                "            )" + MykafkaUtil.getUpsert("dwd_traffic_user_new_user"));

        // TODO 将关联的结果写到kafka主题中
        tableEnv.executeSql("insert into dwd_traffic_user_new_user select * from user_info");

    }
}
